package avicit.bdp.dds.api.utils.notebook;

import avicit.bdp.dds.api.dto.NoteBookDTO;
import avicit.bdp.dds.common.utils.OSUtils;
import avicit.platform6.core.exception.BusinessException;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * @金航数码科技有限责任公司
 * @作者：developer
 * @邮箱：developer@avic-digital.com
 * @创建时间： 2022-05-26
 * @类说明：NoteBookMonitor
 * @修改记录：
 * @注意事项：
 * @主要功能： 1、监控NoteBook后台服务，定时更新状态（MySQL、Redis）
 * 2、实际使用时，仅当DDS重启后会更新状态，正常场景下ProcessBuilder监控后台进程。
 */
@Component
public class NoteBookMonitor {
    private static final Logger logger = LoggerFactory.getLogger(NoteBookMonitor.class);

    /**
     * process
     */
    private Process process;

    /**
     * 从数据库中获取的进程列表
     */
    private Map<Integer, NoteBookDTO> runningNoteBookMap;

    /**
     * 定时任务，每隔5s检查下NoteBook后台服务状态
     * 由于DDS未重启场景下，ProcessBuilder会监控后台进程。只有重启DDS后才需要更新状态、重建缓存关系，所有定时时间不用太短。
     */
    @Scheduled(initialDelay = 10 * 1000, fixedDelayString = "30000")
    public void run() {
        try {
            logger.debug(">>>>>>>>>>>>>>>开启执行NoteBook定时服务<<<<<<<<<<<<<<");
            runCommand();
            logger.debug(">>>>>>>>>>>>>>>执行NoteBook定时服务完成<<<<<<<<<<<<<<");
        } catch (Exception e) {
            throw new BusinessException("NoteBook后台监控任务执行失败,errMsg=" + e.getMessage());
        }
    }

    /**
     * 执行命令
     *
     * @throws Exception
     */
    private void runCommand() throws Exception {
        this.runningNoteBookMap = NoteBookExecuteUtils.getRunningBookNoteList();
        if (this.runningNoteBookMap == null || this.runningNoteBookMap.isEmpty()) {
            return;
        }

        List<String> pidList = getPidList(runningNoteBookMap);
        if (CollectionUtils.isEmpty(pidList)) {
            return;
        }

        List<String> execCommand = new LinkedList<>();
        if (OSUtils.isWindows()) {
            // TODO 目前不支持window环境
            //throw new BusinessException("当前不支持Windows环境.");
            return;
        } else {
            /**
             * 注意1：此处不能使用重定向，ps -p pid | awk 'print $1'，ProcessBuilder不支持输出结果重定向；
             */
            execCommand.add("ps");
            execCommand.add("-p");
            execCommand.add(String.join(" ", pidList));

            logger.debug(">>>>Build Jupyter NoteBook Monitor Process, command = {}", StringUtils.join(execCommand, " "));

            ProcessBuilder processBuilder = new ProcessBuilder();
            processBuilder.redirectErrorStream(true);
            processBuilder.command(execCommand);
            process = processBuilder.start();
        }

        // 获取执行结果
        List<Integer> runningPidList = parseProcessOutput();

        /**
         * 更新缓存；
         * 返回值0：执行成功。
         * 返回值1：操作不允许。但当ps -p pid命令 && pid不存活时，会返回1；所以也认为是正常状态；
         * 其他返回值：致命错误，可以不更新状态、缓存；
         *
         * 结论：不论执行结果，统一更新状态
         */
        int status = process.waitFor();
        updateNoteBooksStatus(runningPidList);
        updateNoteBooksCache(runningPidList);
    }

    /**
     * 获取pid列表
     *
     * @param runningNoteBookMap
     * @return
     */
    private List<String> getPidList(Map<Integer, NoteBookDTO> runningNoteBookMap) {
        List<String> pidList = new ArrayList<>();
        for (Map.Entry<Integer, NoteBookDTO> entry : runningNoteBookMap.entrySet()) {
            if (entry.getKey() != null) {
                pidList.add(entry.getKey().toString());
            }
        }

        return pidList;
    }

    /**
     * 从shell命令执行结果中获取运行的NoteBook服务进程ID列表
     *
     * @return
     */
    private List<Integer> parseProcessOutput() {
        if (process == null) {
            return new ArrayList<>();
        }

        List<Integer> result = new ArrayList<>();
        BufferedReader inReader = null;
        try {
            if (OSUtils.isWindows()) {
                throw new BusinessException("当前不支持Windows环境.");
            } else {
                inReader = new BufferedReader(new InputStreamReader(process.getInputStream(), "UTF-8"));
            }

            String line;
            while ((line = inReader.readLine()) != null) {
                String[] strs = line.trim().split(" ");
                if (StringUtils.isNumeric(strs[0])) {
                    result.add(Integer.parseInt(strs[0]));
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            close(inReader);
        }

        return result;
    }

    /**
     * 更新NoteBook状态
     *
     * @param runningProcessIdList
     */
    private void updateNoteBooksStatus(List<Integer> runningProcessIdList) {
        if (this.runningNoteBookMap == null || this.runningNoteBookMap.isEmpty()) {
            return;
        }

        for (Map.Entry<Integer, NoteBookDTO> entry : this.runningNoteBookMap.entrySet()) {
            if (entry.getKey() == null) {
                continue;
            }
            if (entry.getValue() == null) {
                continue;
            }

            /**
             * 修改说明：数据中notebook记录，由于重启等原因环境中并无notebook进程，Monitor会一直更新db和redis，gate-way监控redis变化
             *         会不断进行业务处理，可能导致异常日志或不必要的错误。
             * 修改时间：2023-11-06 15:10
             * 修改人：wm
             */
            // db中记录的运行状态
            Integer prevStatus = entry.getValue().getStatus();
            // 环境中实际运行状态(能够找到进程，表明notebook正在运行；否则notebook未启动)
            Integer currStatus = runningProcessIdList.contains(entry.getKey()) ? 1 : 0;
            if (currStatus != prevStatus) {
                NoteBookExecuteUtils.updateNoteBookStatus(entry.getValue(), currStatus);
            }

//            if(runningProcessIdList.contains(entry.getKey())) {
//                NoteBookExecuteUtils.updateNoteBookStatus(entry.getValue(), 1);
//            } else {
//                NoteBookExecuteUtils.updateNoteBookStatus(entry.getValue(), 0);
//            }
        }
    }

    /**
     * 重建NoteBook服务缓存信息
     * 注意：dds重启后，对于重启前运行中服务无法监控，导致无法停止。所以，重建缓存关系
     *
     * @param runningPidList
     */
    private void updateNoteBooksCache(List<Integer> runningPidList) {
        if (this.runningNoteBookMap == null || this.runningNoteBookMap.isEmpty()) {
            return;
        }

        for (Integer pid : runningPidList) {
            NoteBookDTO noteBookDTO = this.runningNoteBookMap.get(pid);
            if (noteBookDTO == null) {
                continue;
            }
            if (NoteBookExecuteUtils.getNoteBookTaskContextById(noteBookDTO.getId()) != null) {
                continue;
            }

            NoteBookTaskContext taskContext = new NoteBookTaskContext();
            taskContext.setId(noteBookDTO.getId());
            taskContext.setProcessId(pid);
            NoteBookExecuteUtils.cacheTaskExecutionContext(taskContext);
        }
    }

    /**
     * 关闭流
     *
     * @param inReader
     */
    private void close(BufferedReader inReader) {
        if (inReader != null) {
            try {
                inReader.close();
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }
}
